package service

import (
	"bbs-web/internal/domain"
	"bbs-web/internal/repository"
	"bbs-web/internal/repository/dao"
	"bbs-web/pkg/logger"
	"context"
	"time"
)

type JobService interface {
	// Preempt 抢占
	Preempt(ctx context.Context) (domain.Job, error)
	Release(ctx context.Context, id int64) error
	ResetNextTime(ctx context.Context, j domain.Job) error
}

// preemptCronJobService
// @Description: 抢占式的分布式任务调度 基于MySQL
type preemptCronJobService struct {
	repo            repository.JobRepository
	refreshInterval time.Duration
	l               logger.Logger
}

func NewPreemptCronJobService(repo repository.JobRepository) JobService {
	return &preemptCronJobService{repo: repo}
}

func (p preemptCronJobService) ResetNextTime(ctx context.Context, j domain.Job) error {
	nextTime := j.NextTime()
	if nextTime.IsZero() {
		return p.repo.Release(ctx, j.ID)
	}
	return p.repo.UpdateNextTime(ctx, j.ID, nextTime)
}

// Preempt
//
//	@Description: 抢占式调度，不返回回调函数
func (p preemptCronJobService) Preempt(ctx context.Context) (domain.Job, error) {
	j, err := p.repo.Preempt(ctx)
	if err != nil {
		return domain.Job{}, err
	}
	ticker := time.NewTicker(p.refreshInterval)
	//开协程定期刷新
	go func() {
		// 续约  定时去续约
		for range ticker.C {
			p.refresh(j.ID)
		}
	}()

	// 是否一直抢占？是不是需要释放呢？暴漏release吗
	// 将一个取消函数赋值给domain-job任务结构体的 CancelFunc 字段，用于在需要时释放任务。
	j.CancleFunc = func() error {
		// 这里用来释放锁
		ticker.Stop()
		ctx1, cancel := context.WithTimeout(context.Background(), time.Second)
		defer cancel()
		err = p.Release(ctx1, j.ID)
		if err != nil {
			p.l.Error("释放 job 失败",
				logger.Error(err),
				logger.Int64("job_id", j.ID))
		}
		return err
	}
	return j, err
}

func (p *preemptCronJobService) Release(ctx context.Context, id int64) error {
	return p.repo.Release(ctx, id)
}

// 定期刷新
func (p preemptCronJobService) refresh(id int64) {
	// 续约： 更新一下更新时间即可
	ctx, cancle := context.WithTimeout(context.Background(), time.Second)
	defer cancle()
	err := p.repo.UpdateUpdateTime(ctx, id)
	if err != nil {
		// 续约失败 理论上要通知调用方 续约失败
		p.l.Error("续约失败",
			logger.Error(err),
			logger.Int64("job_id", id))
	}
	// 续约失败判定逻辑: 处于running状态，但是update_time在三分钟之前,说明你没有续约。
	job, err := p.repo.GetJob(ctx, id)
	if err != nil {
		p.l.Error("获取任务信息失败",
			logger.Error(err),
			logger.Int64("job_id", id))
	}
	if job.Status == dao.JobStatusRunning && time.Since(job.UpdatedAt) > 3*time.Minute {
		p.l.Error("续约失败",
			logger.Error(err),
			logger.Int64("job_id", id))
	}
}
